#!/usr/bin/env python3
# coding:utf-8

import base64
import configparser
import datetime
import hashlib
import hmac
import json
import logging
import os
import random  # 调用random、string模块
import string
import sys
import threading
import time
import urllib.parse
from collections import Counter
from concurrent.futures import ThreadPoolExecutor as Pool, as_completed

import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning

requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
from apscheduler.schedulers.background import BackgroundScheduler
from flask import Flask, request, jsonify, abort
from flask_apscheduler import APScheduler
from flask_restful import reqparse, Api, Resource
from kubernetes import client, config
from requests import post, get

pwd = os.path.split(os.path.realpath(__file__))[0]
logging.basicConfig(format="%(asctime)s %(levelname)s %(funcName)s %(message)s", level=logging.INFO, filemode='a',
                    filename=pwd + '/' + 'eventer.log')
kubeversion = os.getenv('KUBEVERSION','1.20.6')
clustername = os.getenv('CLUSTERNAME','测试K8s')
alert_map = {}
exitCodeMap = {
    0: '正常退出,启动时的0表示特定容器没有附加前台进程',
    1: '容器因应用程序错误或镜像规范中的错误引用而停止',
    125: '容器未能运行,docker run命令没有执行成功',
    126: '命令调用错误 无法调用镜像中指定的命令',
    127: '找不到文件或目录	找不到镜像中指定的文件或目录',
    128: '退出时使用的参数无效',
    134: '异常终止 (SIGABRT),容器使用 abort() 函数自行中止',
    137: '立即终止 (SIGKILL)	容器被操作系统通过 SIGKILL 信号强行终止',
    139: '程序试图访问不被允许访问的内存地址，可能是程序代码或者是基础镜像的错误',
    143: '优雅终止 (SIGTERM)	容器收到即将终止的警告',
    255: '退出状态超出范围,错误原因未知'
}


class DateEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime):
            return obj.strftime("%Y-%m-%d %H:%M:%S")
        else:
            return json.JSONEncoder.default(self, obj)


class Config(object):
    JOBS = [
        {
            'id': 'nodejob',
            'func': 'k8s-eventer:nodeconditions',
            'args': '',
            'trigger': 'interval',
            'seconds': 60
        }
    ]
    SCHEDULER_TIMEZONE = 'Asia/Shanghai'
    SCHEDULER_API_ENABLED = True


class KubeConn:

    @staticmethod
    def __conn(uri):
        try:
            host = 'https://'  #k8sapi 地址
            token = ''
            verify_ssl = False
            api_url = f'{host}{uri}'
            api_key = {"Authorization": f"Bearer {token}"}
            response = requests.get(api_url, headers=api_key, verify=verify_ssl)
        except Exception as e:
            print(e, e.__traceback__.tb_lineno)
        else:
            return response.json()

    def list_namespace(self):
        uri = '/api/v1/namespaces'
        namespaces = [item['metadata']['name'] for item in self.__conn(uri)['items'] if
                      item['status']['phase'] == 'Active']
        return namespaces

    def list_namespace_pod(self, namespace):
        uri = f'/api/v1/namespaces/{namespace}/pods'

    def list_node(self):
        uri = f'/api/v1/nodes'
        resp = self.__conn(uri)
        return resp

    def list_pod_for_all_namespaces(self, namespace):
        uri = f'/api/v1/pods'

    def read_namespace_pod(self, namespace, name):
        uri = f'/api/v1/namespaces/{namespace}/pods/{name}'
        resp = self.__conn(uri)
        return podconvert(resp)


def podconvert(p):
    '''
        依次是应用名、命名空间、pod名、podip地址、pod状态、pod所在节点、退出码
    '''
    namespace = p['metadata'].get('namespace')
    container_statuses = p['status'].get('containerStatuses')
    try:
        def port_get(p):
            for containerspec in p['spec']['containers']:
                port = containerspec['livenessProbe'].get('tcpSocket').get('port')
                return port

        def cur_state(**state):
            for k, v in state.items():
                if v is not None:
                    return k, v

        for container in container_statuses:
            restart_count = container.get('restartCount')
            name = container.get('name')
            phase = p['status'].get('phase')
            exitcode = ''
            cur_state = cur_state(**container.get('state'))
            version = str(container.get('image')).split('/')[-1]
            try:
                if list(container['lastState'].keys())[0] == 'terminated':
                    exitcode = container['lastState'].get('terminated').get('exitCode')
            except Exception:
                pass
            return (name, namespace, p['metadata'].get('name'), f"{p['status'].get('podIP')}:{port_get(p)}",
                    p['status'].get('hostIP'), version, phase, restart_count, exitcode)
    except Exception as e:
        logging.error(f'{e} {e.__traceback__.tb_lineno}')


def nodeconditions() -> dict:
    def address_turn(*addlist):
        nodeip = ''
        nodename = ''
        for addr in addlist[0]:
            if addr.get('type') == 'Hostname':
                nodename = addr.get('address')
            elif addr.get('type') == 'InternalIP':
                nodeip = addr.get('address')
        return (nodeip, nodename)

    conditions_dict = {}
    conn = KubeConn().list_node()
    for item in conn.get('items'):
        addresses = item['status'].get('addresses')
        nodeip, nodename = address_turn(addresses)
        conditions = item['status'].get('conditions')
        filter_conditions = filter(lambda x: (x.get('status') == 'True' and x.get('reason') != 'KubeletReady') or (
                    x.get('status') == 'False' and x.get('reason') == 'KubeletReady'), conditions)
        conditions_dict[nodename] = list(filter_conditions)
    alert().handle_condition(**conditions_dict)


class alert:
    _url = "https://oapi.dingtalk.com/robot/send?access_token="  # 告警地址
    interval = 1  # 告警的静默时间，单位小时

    @staticmethod
    def get_sign():
        timestamp = str(round(time.time() * 1000))
        secret = ''  #钉钉密钥
        secret_enc = secret.encode('utf-8')
        string_to_sign = timestamp + '\n' + secret
        string_to_sign_enc = string_to_sign.encode('utf-8')
        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
        return timestamp, sign

    def send_msg(self, title, append='', **all_events):
        ns = f"{all_events['EventNamespace']}/" if all_events.get('EventNamespace') else ''
        data = {
            "msgtype": "markdown",
            "markdown": {
                "title": title,
                "text": "<font color=#FF0000 size=3>事件详情：</font>" +
                        "\n\n>事件时间：" + str(all_events['EventTime']) +
                        "\n\n>应用名称：" + f"{ns}" + str(all_events['EventObject']) +
                        "\n\n>事件类型：" + str(all_events['EventReason']) +
                        f"\t({all_events['EventCount']}counts)" +
                        "\n\n>事件详情：" + str(all_events['EventMessage']) + append
            },
            "at": {
                "atMobiles": [
                    "123456"
                ]
            },
            "isAtAll": "False"
        }
        timestamp, sign = self.get_sign()
        params1 = {"timestamp": timestamp, "sign": sign}
        r = post(self._url, json=data, params=params1)
        return r.json()

    def send_conditions_msg(self, title, nodename, **all_events):
        try:
            logging.info(f'{nodename} 触发 condition 事件 {all_events}')
            assert self.__filter_condition(all_events.get('reason'))
            data = {
                "msgtype": "markdown",
                "markdown": {
                    "title": title,
                    "text": "<font color=#FF0000 size=3>事件详情：</font>" +
                            "\n\n>触发主机：" + str(nodename) +
                            "\n\n>触发时间：" + str(all_events['lastTransitionTime']) +
                            "\n\n>触发类型：" + str(all_events['type']) +
                            "\n\n>触发原因：" + str(all_events['reason']) +
                            "\n\n>触发详情：" + str(all_events['message'])
                },
                "at": {
                    "atMobiles": [
                        "123456"
                    ]
                },
                "isAtAll": "False"
            }
            timestamp, sign = self.get_sign()
            params1 = {"timestamp": timestamp, "sign": sign}
            r = post(self._url, json=data, params=params1)
            return r.json()
        except AssertionError:
            # logging.info(f'no alert condition {all_events}')
            pass

    @staticmethod
    def __filter_event(EventReason, Count):
        # 过滤一些告警
        filter_list = ['EnsureNginxIngress', 'Unhealthy', 'CgroupOOM']
        if EventReason in filter_list and int(Count) < 5 or int(Count) > 100:
            return False
        else:
            return True

    @staticmethod
    def __filter_condition(reason):
        filter_list = ['DockerHung']
        if reason in filter_list:
            return False
        else:
            return True

    def handle_event(self, **data):
        title = f"{clustername} {data.get('EventKind')}事件通知"
        Object = data.get('EventObject')
        now = datetime.datetime.now()
        try:
            assert self.__filter_event(data.get('EventReason'), data.get('EventCount'))
            al_time = alert_map.get(Object)
            if al_time:
                if al_time > now:
                    logging.info(f'{Object} 静默中 {al_time}')
                    pass
                else:
                    self.alerts_silent(Object, now, title, **data)
            else:
                self.alerts_silent(Object, now, title, **data)
        except AssertionError:
            logging.info(f"no alert {Object} {data.get('EventMessage')} {data.get('EventCount')}")
        except Exception as e:
            logging.error(f'{e} {e.__traceback__.tb_lineno} {data}')

    def handle_condition(self,**data):
        for k, v in data.items():
            if len(v) > 0:
                global clustername
                title = f'{clustername} Node Condition通知'
                [self.send_conditions_msg(title, k, **vd) for vd in v]


    def alerts_silent(self, Object, now, title, **data):
        global alert_map
        try:
            if data.get('EventKind') == 'Pod':
                podinfo = KubeConn().read_namespace_pod(data.get('EventNamespace'), Object)
                append = f"\n\n>应用地址：{podinfo[3]}\n\n>版本信息：{podinfo[5]}" if podinfo[
                                                                                -1] is '' else f"\n\n>应用地址：{podinfo[3]}\n\n>版本信息：{podinfo[5]}\n\n>退出码：{podinfo[-1]} {exitCodeMap.get(podinfo[-1])}"
                self.send_msg(title=title, append=append, **data)
            else:
                self.send_msg(title=title, **data)
        except Exception as e:
            logging.error(f'{e} {e.__traceback__.tb_lineno}')
        logging.info(f'{Object} 静默 {self.interval}h')
        alert_map[Object] = now + datetime.timedelta(hours=self.interval)


parser = reqparse.RequestParser()
parser.add_argument('EventType', type=str, location='json')
parser.add_argument('EventKind', type=str, location='json')
parser.add_argument('EventObject', type=str, location='json')
parser.add_argument('EventReason', type=str, location='json')
parser.add_argument('EventTime', type=str, location='json')
parser.add_argument('EventMessage', type=str, location='json')


class Eventer(Resource):
    def get(self):
        return jsonify({'status': 'success'})

    def post(self):
        try:
            # args = parser.parse_args()
            # args_dict = {k: v for k, v in args.items()}
            args_dict = eval(request.get_data().decode('utf-8'))
            # logging.info(args_dict)
        except SyntaxError:
            args_dict = {}  # 部分value 中含有引号直接转换有问题的,使用下面的方法转换
            for item in request.get_data().decode('utf-8').strip('{|}').split(','):
                i = item.partition(':')
                k, v = str(i[0]).replace('"', ''), str(i[-1]).replace('"', '')
                args_dict.update({k: v})
        except Exception as e:
            logging.error(f'{e} {e.__traceback__.tb_lineno} {request.get_data()}')
            return abort(400)
        else:
            t = threading.Thread(target=alert().handle_event, kwargs=args_dict)  # 异步线程
            t.start()
            return jsonify({'status': 'success'})


app = Flask(__name__)
api = Api(app)
api.add_resource(Eventer, '/api/eventer/handler')

if __name__ == '__main__':
    app.config.from_object(Config())
    scheduler = APScheduler(BackgroundScheduler(timezone="Asia/Shanghai"))
    scheduler.init_app(app)
    scheduler.start()
    app.run(host="0.0.0.0", port=8081, debug=False)
